package com.gzyj.flink.gps.map.fatiguedriving;

import cn.hutool.core.lang.Pair;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.bestvike.linq.Linq;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.dto.alarm.Kafka_VehicleRemindContent;
import com.gzyj.flink.dto.alarm.RemindContent;
import com.gzyj.flink.dto.alarm.TotalFatigueDrivingData;
import com.gzyj.flink.dto.utils.ToolUtils;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.mapper.mysql.MonPilaoSetMapper;
import com.gzyj.flink.gps.mapper.mysql.TsAlarmRuleMapper;
import com.gzyj.flink.gps.service.NormalAlarmService;
import com.gzyj.flink.gps.utils.TimeUtils;
import com.gzyj.flink.pojo.mysql.set.MonPilaoSet;
import com.gzyj.flink.pojo.mysql.set.TsAlarmRule;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.util.Date;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class TotalFatigueDrivingDataProcess extends RichFlatMapFunction<GPSPosition, FatigueDrivingResult> {

    LettuceConnectionFactory redisFactory;

    SqlSessionFactory mysql;

    NormalAlarmService normalAlarmService;

    private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisSessionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        redisFactory = RedisSessionFactory.getRedisFactory();
        mysql = MySQLMybatisSessionFactory.getSqlSessionFactory();
        normalAlarmService = new NormalAlarmService();
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    private TotalFatigueDrivingData TryGetData_TFDD(String vehicleId, RedisTemplate<String, Object> clusterRedisTemplate) {
        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_TotalFatigueDriving + vehicleId);
        if (obj != null) {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), TotalFatigueDrivingData.class);
        } else {
            return new TotalFatigueDrivingData();
        }
    }

    public Pair<Boolean, MonPilaoSet> GetRuleSetting(GPSPosition gpsPosition, MonPilaoSetMapper monPilaoSetMapper, TsAlarmRuleMapper tsAlarmRuleMapper) {
        MonPilaoSet setting = monPilaoSetMapper.selectByVehID(gpsPosition.getVehId());
        if (setting != null) {
            return new Pair<>(true, setting);
        } else {
            TsAlarmRule rule = tsAlarmRuleMapper.selectByRuleTypePilao(gpsPosition.getShiyongxingzhi());
            if (rule == null || rule.getRuleFlag() == 0) {
                return new Pair<>(false, setting);
            } else {
                setting = monPilaoSetMapper.selectByVehID(gpsPosition.getVehId());
                if (setting != null) {
                    return new Pair<>(true, setting);
                } else {
                    return new Pair<>(false, setting);
                }
            }
        }
    }

    public Driverlogininfo TryGetData_DLD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {
        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_DriverSignOn + vehId);
        if (obj == null) {
            return new Driverlogininfo();
        } else {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Driverlogininfo.class);
        }

    }

    private RemindContent GenerateVehicleRemind(TotalFatigueDrivingData data, MonPilaoSet setting, GPSPosition gps, String msgType) {

        if (msgType.equalsIgnoreCase("预警短信")) {
            RemindContent content = new RemindContent() {
                {
                    VehicleId = gps.getVehId();
                    Ipaddress = gps.getTerminalPhoneNo();
                    VehicleNo = gps.getCheliangpaizhao();
                    PlateColor = gps.getChepaiyanse();
                    RemindNote = setting.getLjpljsyjtxnr();
                    RemindType = 0x01;
                    RemindTime = new Date();
                    MessageType = "累计疲劳驾驶报警";
                    AlarmNote = "累计疲劳驾驶报警";
                    AlarmGuid = data.SingleGuid;
                    GpsTime = data.GpsTime;
                    LastSpeed = gps.getSpeedExt();
                }

            };
            return content;
        } else if (msgType.equalsIgnoreCase("报警短信")) {
            RemindContent content = new RemindContent() {

                {
                    VehicleId = gps.getVehId();
                    Ipaddress = gps.getTerminalPhoneNo();
                    VehicleNo = gps.getCheliangpaizhao();
                    PlateColor = gps.getChepaiyanse();
                    RemindNote = setting.getLjplbjnr();
                    RemindType = 0x02;
                    RemindTime = new Date();
                    MessageType = "累计疲劳驾驶报警";
                    AlarmNote = "累计疲劳驾驶报警";
                    AlarmGuid = data.SingleGuid;
                    GpsTime = data.GpsTime;
                    LastSpeed = gps.getSpeedExt();
                }

            };
            return content;
        } else {
            return null;
        }

    }

    private Kafka_VehicleRemindContent DisposeAlarmRemind(TotalFatigueDrivingData fatigueDrivingData, MonPilaoSet mySetting, GPSPosition gps, String msgType) {
        if (msgType.equalsIgnoreCase("预警短信")) {
            if (mySetting.getLjpljsyjenable() == 1 && mySetting.getLjpljsyjtxcs() > 0 && StringUtils.isNotEmpty(mySetting.getLjpljsyjtxnr())) {
                long ts = TimeUtils.getMinusSecond(new Date(), fatigueDrivingData.WarningContentSendTime);
                if (ts >= mySetting.getLjpljsyjtxjg() && fatigueDrivingData.WarningContentCount < mySetting.getLjpljsyjtxcs()) {
                    fatigueDrivingData.WarningContentCount++;
                    fatigueDrivingData.WarningContentSendTime = new Date();
                    RemindContent remindContent = GenerateVehicleRemind(fatigueDrivingData, mySetting, gps, msgType);
                    return normalAlarmService.SendRemindToKafka(remindContent);
                }
            }
        } else if (msgType.equalsIgnoreCase("报警短信")) {
            if (mySetting.getLjplbjcs() > 0 && StringUtils.isNotEmpty(mySetting.getLjplbjnr())) {
                long ts = TimeUtils.getMinusSecond(new Date(), fatigueDrivingData.AlarmContentSendTime);
                if (ts >= mySetting.getLjplbjjg() && fatigueDrivingData.AlarmContentCount < mySetting.getLjplbjcs()) {
                    fatigueDrivingData.AlarmContentCount++;
                    fatigueDrivingData.AlarmContentSendTime = new Date();
                    RemindContent remindContent = GenerateVehicleRemind(fatigueDrivingData, mySetting, gps, msgType);
                    return normalAlarmService.SendRemindToKafka(remindContent);
                }
            }
        }
        return null;
    }

    @Override
    public void flatMap(GPSPosition gpsPosition, Collector<FatigueDrivingResult> collector) throws Exception {

        RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);
        SqlSession sqlSession = mysql.openSession(true);

        try {
            TsAlarmRuleMapper tsAlarmRuleMapper = sqlSession.getMapper(TsAlarmRuleMapper.class);
            MonPilaoSetMapper monPilaoSetMapper = sqlSession.getMapper(MonPilaoSetMapper.class);


            Driverlogininfo driverLoginData = TryGetData_DLD(gpsPosition.getVehId(), clusterRedisTemplate);


            TotalFatigueDrivingData totalFatigueDrivingData = TryGetData_TFDD(gpsPosition.getVehId(), clusterRedisTemplate);
            if (totalFatigueDrivingData == null) return;
            Pair<Boolean, Integer> p = TimeUtils.CheckData(totalFatigueDrivingData.getGpsTime(), gpsPosition.getTIME());

            int i = p.getValue();

            if (!p.getKey()) {
                return;
            } else {
                totalFatigueDrivingData.LocalTime = new Date();
                totalFatigueDrivingData.GpsTime = gpsPosition.getGPSTimeExt();
            }
            boolean isAlarming = totalFatigueDrivingData.IsAlarm;
            //获取运算参数
            Pair<Boolean, MonPilaoSet> p1 = GetRuleSetting(gpsPosition, monPilaoSetMapper, tsAlarmRuleMapper);
            MonPilaoSet mySetting = p1.getValue();
            if (p1.getKey()) {
                //如果车辆配置了规则

                if (mySetting.getEnable() == 1 && mySetting.getLjplenable() == 1) {
                    //记录车辆的计算数据
                    if (!totalFatigueDrivingData.OnceDayTotalRunTime.containsKey(TimeUtils.getToday())) {
                        //如果跨天后，重置。
                        totalFatigueDrivingData.GenerateSingleGuid();
                        totalFatigueDrivingData.ResetData();
                    }
                    if (!totalFatigueDrivingData.IsCalc) {
                        //开始计算前，重置。
                        totalFatigueDrivingData.GenerateSingleGuid();
                        totalFatigueDrivingData.ResetData();
                        totalFatigueDrivingData.StartCalc(gpsPosition);
                        String note = "开始计算";
                    }
                    double mile = ToolUtils.GetDistance(gpsPosition.getLngExt(), gpsPosition.getLatExt(), totalFatigueDrivingData.ConditionEndLongitude, totalFatigueDrivingData.ConditionEndLatitude);
                    long seconds = TimeUtils.getMinusSecond(gpsPosition.getGPSTimeExt(), totalFatigueDrivingData.ConditionEndTime);
                    if (mile > 0 && seconds <= 5 * 60 && totalFatigueDrivingData.ConditionEndSpeed > 0 && gpsPosition.getSpeedExt() > 0) {
                        totalFatigueDrivingData.OnceDayTotalRunTime.put(TimeUtils.getToday(), totalFatigueDrivingData.OnceDayTotalRunTime.get(TimeUtils.getToday()) + seconds);
                    }
                    totalFatigueDrivingData.WriteCalc(gpsPosition);
                    Map.Entry<String, Double> first = Linq.of(totalFatigueDrivingData.OnceDayTotalRunTime).firstOrDefault();
                    double onceDayTotalRunTime = first.getValue();
                    if (onceDayTotalRunTime > mySetting.getLjplzcjssj() * 60 * 60) {
                        //车辆满足累计驾驶超过8小时
                        if (!totalFatigueDrivingData.IsAlarm) {
                            //开始报警
                            totalFatigueDrivingData.StartAlarm(gpsPosition);

                            //从没报警切换到报警，发送开始报警

                            collector.collect(new FatigueDrivingResult(
                                    normalAlarmService.SendAlarmToKafka(totalFatigueDrivingData, gpsPosition, driverLoginData, true),
                                    DisposeAlarmRemind(totalFatigueDrivingData, mySetting, gpsPosition, "报警短信")
                            ));
                            //特殊处理，报警开始即结束
                            totalFatigueDrivingData.StopAlarm(gpsPosition);
                            //发送停止报警
                            collector.collect(new FatigueDrivingResult(
                                    normalAlarmService.SendAlarmToKafka(totalFatigueDrivingData, gpsPosition, driverLoginData, false),
                                    null
                            ));
                            totalFatigueDrivingData.GenerateSingleGuid();
                            totalFatigueDrivingData.ResetData();
                        }
                    } else {
                        //车辆不满足报警条件
                        if (onceDayTotalRunTime > mySetting.getLjplzcjssj() * 60 * 60 - mySetting.getLjpljsyjtqsj() * 60 && onceDayTotalRunTime <= mySetting.getLjplzcjssj() * 60 * 60) {

                            collector.collect(new FatigueDrivingResult(
                                    null,
                                    DisposeAlarmRemind(totalFatigueDrivingData, mySetting, gpsPosition, "预警短信")
                            ));
                        }
                    }
                } else {
                    //如果车辆规则关闭
                    if (totalFatigueDrivingData.IsAlarm) {
                        totalFatigueDrivingData.StopAlarm(gpsPosition);
                        //发送停止报警
                        collector.collect(new FatigueDrivingResult(
                                normalAlarmService.SendAlarmToKafka(totalFatigueDrivingData, gpsPosition, driverLoginData, false), null
                        ));
                    }
                    totalFatigueDrivingData.GenerateSingleGuid();
                    totalFatigueDrivingData.ResetData();
                }
            } else {
                //如果车辆没有配置规则或者取消了规则
                if (totalFatigueDrivingData.IsAlarm) {
                    totalFatigueDrivingData.StopAlarm(gpsPosition);
                    //发送停止报警
                    collector.collect(new FatigueDrivingResult(
                            normalAlarmService.SendAlarmToKafka(totalFatigueDrivingData, gpsPosition, driverLoginData, false), null
                    ));
                }
                totalFatigueDrivingData.GenerateSingleGuid();
                totalFatigueDrivingData.ResetData();
            }

            clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_TotalFatigueDriving + gpsPosition.getVehId(), totalFatigueDrivingData, 8, TimeUnit.HOURS);


        } catch (Exception e) {
            e.printStackTrace();
            sqlSession.rollback();
        } finally {
            sqlSession.close();
        }

    }
}
